package com.iot.service.mqtt.service.pub;

import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.service.StoreStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: hszhou
 * @Date: 2025/4/18 14:46
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
@Slf4j
public class LocalPublishStoreServiceImpl implements IPublishStoreService {


    public static final ConcurrentHashMap<String, ConcurrentHashMap<String, TlMqttPublishReq>> PUBLISH_MAP = new ConcurrentHashMap<>();


    public static final ConcurrentHashMap<String,TlMqttPublishReq> WILL_MAP = new ConcurrentHashMap<>();

    @Override
    public Mono<TlMqttPublishReq> savePublishMessage(String clientId, Long messageId, TlMqttPublishReq req) {

        return Mono.fromSupplier(()->
                PUBLISH_MAP.compute(clientId, (k, v) -> {
                    if (v == null) {
                        v = new ConcurrentHashMap<>(16);
                    }
                    v.computeIfAbsent(String.valueOf(messageId), key -> req);
                    return v;
                })).thenReturn(req);

    }

    @Override
    public Mono<TlMqttPublishReq> clearPublishMessage(String clientId, Long messageId) {
        return Mono.fromSupplier(()-> PUBLISH_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(16))
                  .remove(String.valueOf(messageId)));
    }

    @Override
    public Mono<Boolean> clearPublishMessage(String clientId) {
        return Mono.fromSupplier(()-> PUBLISH_MAP.remove(clientId)==null);
    }

    @Override
    public Mono<TlMqttPublishReq> findPublishMessage(String clientId, Long messageId) {
        return Mono.fromSupplier(()->PUBLISH_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(10)).get(String.valueOf(messageId)));
    }

    @Override
    public Flux<TlMqttPublishReq> findPublishMessage(String clientId) {

        return Flux.create(sink -> PUBLISH_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(16)).forEach((k, v) -> sink.next(v)));
    }

    @Override
    public Mono<Boolean> saveWillMessage(String clientId, TlMqttPublishReq req) {
       return Mono.defer(()-> WILL_MAP.putIfAbsent(clientId, req) == null ? Mono.just(true) : Mono.just(false));
    }

    @Override
    public Mono<TlMqttPublishReq> findWillMessage(String clientId) {
        return Mono.defer(()->Mono.justOrEmpty(WILL_MAP.get(clientId)));
    }

    @Override
    public Mono<Boolean> clearWillMessage(String clientId) {
        return Mono.defer(()->Mono.just(WILL_MAP.remove(clientId) != null));
    }
}
